package io.xxx.omni.sync

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper
import io.xxx.omni.util.dateTimeFormatter
import io.xxx.omni.util.numberFormat
import org.quartz.*
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.ApplicationArguments
import org.springframework.boot.ApplicationRunner
import org.springframework.boot.web.client.RestTemplateBuilder
import org.springframework.context.ApplicationContext
import org.springframework.core.io.Resource
import org.springframework.http.HttpEntity
import org.springframework.http.HttpHeaders
import org.springframework.http.HttpMethod
import org.springframework.http.client.OkHttp3ClientHttpRequestFactory
import org.springframework.scheduling.quartz.QuartzJobBean
import org.springframework.stereotype.Component
import org.springframework.util.LinkedMultiValueMap
import org.springframework.web.client.HttpServerErrorException
import org.springframework.web.client.RestTemplate
import java.io.BufferedReader
import java.io.InputStreamReader
import java.net.URI
import java.time.Duration
import java.time.LocalDateTime
import java.util.*
import java.util.concurrent.*
import kotlin.collections.ArrayList

@Component
class Synchronizer : ApplicationRunner {

    @Autowired
    private lateinit var platformMapper: PlatformMapper

    @Autowired
    private lateinit var applicationContext: ApplicationContext

    @Autowired
    private lateinit var scheduler: Scheduler

    /**
     * key: platform code
     * value: porter list
     */
    private val porters = LinkedMultiValueMap<String, Porter>()

    override fun run(args: ApplicationArguments) {
        fun getPlatforms(): List<Platform> {
            return platformMapper.selectList(null)
        }

        val platforms = getPlatforms().associateBy { it.code!! }
        val porters = applicationContext.getBeansOfType(Porter::class.java)
        porters.forEach { (_, porter) ->
            val platform = platforms[porter.platformCode]
            if (platform == null) {
                log.warn("[${porter.platformCode}]平台信息不存在")
                return
            }
            this.porters.add(platform.code!!, porter)
            val className = porter::class.java.simpleName
            val jobDetail = JobBuilder.newJob(PorterHandler::class.java)
                .withIdentity("${className}Job", "SYNC")
                .withDescription("${platform.name}任务")
                .usingJobData(JobDataMap(mapOf("platform" to platform, "porter" to porter)))
                .build()
            val trigger = TriggerBuilder.newTrigger()
                .withIdentity("${className}Trigger", "SYNC")
                .withDescription("${platform.name}触发器")
                .withSchedule(CronScheduleBuilder.cronSchedule(porter.cron))
                .build()
            if (!scheduler.checkExists(jobDetail.key)) {
                scheduler.scheduleJob(jobDetail, trigger)
            }
        }
    }

    companion object {
        private val log: Logger = LoggerFactory.getLogger(Synchronizer::class.java)
    }
}

@DisallowConcurrentExecution
class PorterHandler : QuartzJobBean() {

    override fun executeInternal(context: JobExecutionContext) {
        val dataMap = context.mergedJobDataMap
        (dataMap["porter"] as Porter).pullAndSave()
    }
}

/**
 * Stream porter
 */
abstract class Porter {

    @Autowired
    protected lateinit var jobMapper: JobMapper

    @Autowired
    protected lateinit var platformMapper: PlatformMapper

    @Autowired
    protected lateinit var storeMapper: StoreMapper

    @Autowired
    protected lateinit var storeJobMapper: StoreJobMapper

//    @Autowired
//    protected lateinit var scheduleMapper: ScheduleMapper

    @Autowired
    protected lateinit var documentMapper: DocumentMapper

    @Autowired
    private lateinit var restTemplateBuilder: RestTemplateBuilder

//    @Autowired
//    private lateinit var client: PulsarClient

//    private val producer: Producer<Long> by lazy {
//        client.newProducer(Schema.INT64)
//            .topic("DOCUMENT_CREATED")
//            .sendTimeout(1, TimeUnit.MINUTES)
//            .create()
//    }

    protected val restTemplate: RestTemplate by lazy {
        restTemplateBuilder
            .setConnectTimeout(Duration.ofMinutes(1))
            .setReadTimeout(Duration.ofMinutes(10))
            .requestFactory(OkHttp3ClientHttpRequestFactory::class.java)
            .build()
    }

    protected val container: MutableMap<String, Any> = Collections.synchronizedMap(mutableMapOf<String, Any>())

    /**
     * Platform code
     */
    abstract val platformCode: String

    /**
     * 任务ID，如果平台接口不区分订单和退单时，统一填写订单任务ID
     * @see Job
     */
    abstract val jobId: Int

    private val job by lazy {
        jobMapper.selectById(jobId)
    }

    /**
     * 限制开始时间和结束时间的时间范围不超过平台的限制
     */
    protected open val duration: Duration? = null

    open val cron = "0/5 * * * * ?"

    fun pullAndSave() {
        fun getPlatform(): Platform? {
            val wrapper = QueryWrapper<Platform>()
                .eq("code", platformCode)
                .eq("enabled", true)
            return platformMapper.selectOne(wrapper)
        }

        fun getStores(platform: Platform): List<Store> {
            val wrapper = QueryWrapper<Store>()
                .eq("platform_code", platform.code)
                .eq("enabled", true)
            return storeMapper.selectList(wrapper)
        }

        fun getStoreJob(store: Store): StoreJob? {
            val wrapper = QueryWrapper<StoreJob>()
                .eq("store_code", store.code)
                .eq("job_id", jobId)
            val storeJob = storeJobMapper.selectOne(wrapper)
            if (storeJob.endTime == null) {
                storeJob.endTime = store.created
                storeJobMapper.updateById(storeJob)
            }
            return storeJob
        }

        /**
         * 数据同步需要一对开始时间和结束时间，根据平台对于开始时间和结束时间的时间范围约束生成
         */
        fun getTimeRanges(startTime: LocalDateTime, endTime: LocalDateTime): Set<Pair<LocalDateTime, LocalDateTime>> {
            val timeRanges = mutableSetOf<Pair<LocalDateTime, LocalDateTime>>()
            if (duration == null) {
                timeRanges.add(startTime to endTime)
            } else {
                val totalSeconds = Duration.between(startTime, endTime).abs().seconds
                val indexes = totalSeconds / duration!!.seconds + if (totalSeconds % duration!!.seconds == 0L) 0 else 1
                for (i in 0 until indexes) {
                    val first = startTime.plusSeconds(duration!!.seconds * i)
                    val second = if ((i + 1) == indexes) endTime
                    else startTime.plusSeconds(duration!!.seconds * (i + 1))
                    timeRanges.add(first to second)
                }
            }
            return timeRanges
        }

        val platform = getPlatform() ?: return
        val stores = getStores(platform)
        stores.parallelStream().forEach { store ->
            store.platform = platform

            val storeJob = getStoreJob(store)
            if (!job.enabled!! || !storeJob?.enabled!!) {
                return@forEach
            }

            val startTime = getStartTime(storeJob)
            val endTime = LocalDateTime.now()
            val totalSeconds = Duration.between(startTime, endTime).abs().seconds
            if (totalSeconds < 1) {
                return@forEach
            }

            val timeRanges = getTimeRanges(startTime, endTime)
            timeRanges.forEach TimeRange@{
                val newStoreJob = storeJobMapper.selectById(storeJob.id)
                if (!newStoreJob.enabled!!) {
                    return@TimeRange
                }
                val newStore = storeMapper.selectById(storeJob.storeCode)
                if (!newStore.enabled!!) {
                    return@TimeRange
                }
                val start = System.currentTimeMillis()
                val parameters = if (getParameters(store).isEmpty()) listOf(null)
                else getParameters(store)
                val set = mutableSetOf<String>()
                for (parameter in parameters) {
                    acquire()
                    set.addAll(pullAndSave(store, it.first, it.second, parameter))
                }
                storeJobMapper.updateById(storeJob.copy(enabled = null, endTime = it.second))
                val st = it.first.format(dateTimeFormatter)
                val et = it.second.format(dateTimeFormatter)
                val spend = numberFormat.format((System.currentTimeMillis() - start) / 1000.0)
                log.info("${this::class.java.simpleName}: ${store.name}(${store.code})[${st} - ${et}]同步${set.size}条数据完成，耗时 $spend 秒")
            }
        }

        container.clear()
    }

    /**
     * 进行限流等操作
     */
    protected open fun acquire() {
    }

    /**
     * 使用不同的参数多次调用接口
     */
    protected open fun getParameters(store: Store): List<Any> {
        return emptyList()
    }

    /**
     * 返回麻烦平台限制的开始时间
     */
    protected open fun getStartTime(storeJob: StoreJob): LocalDateTime {
        return storeJob.endTime!!
    }

    protected open fun pullAndSave(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?
    ): List<String> {
        val resource = getResource(store, startTime, endTime, parameter) ?: return emptyList()
        val br = BufferedReader(InputStreamReader(resource.inputStream), 1024 * 1024 * 20)

        val result = CopyOnWriteArrayList<String>()
        val bufferSize = 1_000
        val buffer = ArrayList<String>(bufferSize)
        val futures = mutableListOf<Future<*>>()
        br.lines().forEach { it ->
            buffer.add(it)
            if (buffer.size == bufferSize) {
                val data = ArrayList(buffer)
                val documents = buildDocuments(data)
                result += documents.map { it.sn }
                futures += executors.submit { save(store, documents) }
                buffer.clear()
            }
        }
        if (buffer.isNotEmpty()) {
            val documents = buildDocuments(buffer)
            result += documents.map { it.sn }
            save(store, documents)
        }
        futures.forEach { it.get() }
        return result
    }

    /**
     * 流式查询的接口返回Resource
     */
    protected open fun getResource(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?
    ): Resource? {
        val url = getUrl(store, startTime, endTime, parameter)
        val requestEntity = HttpEntity(null, getHeaders())
        val response = restTemplate.exchange(url, HttpMethod.GET, requestEntity, Resource::class.java)
        if (response.statusCode.isError) {
            throw HttpServerErrorException(response.statusCode, "[${store.platform?.code}]接口异常")
        }
        return response.body
    }

    protected open fun getUrl(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?
    ): URI {
        throw NoSuchMethodException("Not support")
    }

    protected open fun getHeaders(): HttpHeaders? {
        return null
    }

    protected open fun buildDocuments(data: List<String>): List<Document> {
        throw NoSuchMethodException("Not support")
    }

    /**
     * 保存数据
     */
    protected open fun save(store: Store, data: List<Document>?) {
        if (data.isNullOrEmpty()) {
            return
        }
        data.forEach {
            it.platformCode = store.platform!!.code
            if (it.storeCode == null) it.storeCode = store.code
            it.jobId = jobId
        }
        val existed = if (data.isNotEmpty()) {
            val wrapper = QueryWrapper<Document>()
                .select("sn", "modified")
                .eq("store_code", store.code)
                .`in`("sn", data.map { it.sn })
            documentMapper.selectList(wrapper).associateBy { it.sn }
        } else {
            emptyMap()
        }
        val (new, old) = data.partition { !existed.contains(it.sn) }
        val insert = executors.submit {
            if (new.isNotEmpty()) {
                documentMapper.insertAll(new)
//                    new.forEach { producer.send(it.id) }
            }
        }
        val update = executors.submit {
            if (old.isNotEmpty()) {
                val updating = old.filter {
                    val doc = existed[it.sn]!!
                    doc.modified == null || (it.modified!! > existed[it.sn]!!.modified)
                }
                if (updating.isNotEmpty()) {
                    documentMapper.updateAll(updating)
//                        updating.forEach { producer.send(it.id) }
                }
            }
        }
        insert.get()
        update.get()
    }

    companion object {
        private val log: Logger = LoggerFactory.getLogger(Porter::class.java)

        private val executors = ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2,
            120, TimeUnit.SECONDS,
            LinkedBlockingQueue(500)
        )

        fun throwException(api: String?, code: Any?, msg: String?) {
            val message = String.format("%s: %s[%s]", api, code, msg)
            throw RuntimeException(message)
        }
    }
}

/**
 * Page porter
 */
abstract class PagePorter : Porter() {

    protected open val startPage = 1
    protected open val pageSize = 100

    override fun pullAndSave(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?
    ): List<String> {
        val count = getCount(store, startTime, endTime, parameter) ?: 0
        var pages = count / pageSize + if (count % pageSize == 0L) 0 else 1
        val result = mutableListOf<String>()
        while (pages-- > 0) {
            val data = getData(store, startTime, endTime, parameter, pages + startPage)
            if (!data.isNullOrEmpty()) {
                result.addAll(data.map { document -> document.sn!! })
            }
            save(store, data)
        }
        return result
    }

    /**
     * 返回数据总数
     */
    protected open fun getCount(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?
    ): Long? {
        throw NoSuchMethodException("Not support")
    }

    /**
     * 根据分页信息返回每页数据，分页由后向前索引
     */
    protected open fun getData(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?,
        pageNo: Long,
    ): List<Document>? {
        throw NoSuchMethodException("Not support")
    }
}